/* Include files. */

extern "C" { 
	#include "standard.h"
	#include "util.h"
	#include "debug.h"
} 
#include "network.h"
#include "netobject.h"
#include "subsystemcore.h"


/* Local function declarations. */

/* Macro API definitions. */

#define MAX_QUEUE 1000

/* Global variable declarations. */

/* Type definitions. */

/* Local function definitions. */

#ifdef U_LOOP
static void evDealMsg(struct uloop_fd *h, unsigned int events)
{
	netObject *netobj = (netObject *)h->data;
	int ret ;

	if(events & ULOOP_READ) {
		while(e_CONTINUE == (ret = netobj->recvMsg())) ;

		if(ret == e_ERROR)
			goto error;
	}

	if(events & ULOOP_WRITE) {
		while(netobj->_out_queue.size()) {
			ret = netobj->sendMsg();
			switch(ret) {
			case e_ERROR:
				break;
			case e_CONTINUE:
				break;
			case e_OK:
				break;
			}
		}
	}

	return;

error:
	delete netobj;
	return;
}
#endif

int netObject::netRecv(int fd, u8 *buf, int len)
{
	int readed = 0;
	int ret;

	do {
		ret = recv(fd, &buf[readed], len - readed, 0);

		if(ret > 0) {
			readed += ret;
		} else if (0 == ret) {
			readed = -1;
			break;
		} else {
			if (TCP_NORMAL_ERR(errno)) {
				break;
			} else {
				readed = -1;
				break;
			}
		}
	} while(readed < len);

	return readed;
}

int netObject::netSend(int fd, u8 *buf, int len)
{
	int writed = 0;
	int ret;

	do {
		ret = send(fd, &buf[writed], len - writed, MSG_NOSIGNAL);

		if(ret > 0) {
			writed += ret;
		} else if(0 == ret) {
			break;
		} else {
			if(TCP_NORMAL_ERR(errno)) {
				break;
			} else {
				writed = -1;
				break;
			}
		}
	} while(writed < len);

	return writed;
}


netObject::netObject(u32 ip, u16 port, u32 fd)
		:_ip(ip), _port(port), _fd(fd), _id(0), _sub(NULL)
{
	SOCKOPT_NOBLOCK(_fd, 1);
#ifdef U_LOOP
	memset(&_h, 0, sizeof(_h));
	_h.cb = evDealMsg;
	_h.fd = fd;
	_h.data = this;
	uloop_fd_add(&_h, ULOOP_READ | ULOOP_WRITE | ULOOP_EDGE_TRIGGER);
#endif
}

netObject::~netObject() 
{
	DEBUG("close connection " NIPQUAD_FMT ":%d , netobj = %p, socket = %d\n",
	  NIPQUAD(_ip), _port, this, _fd);
#ifdef U_LOOP	
	uloop_fd_delete(&_h);
#else
	if (ev_is_active(&_h)) {
		ev_io_stop(_loop, &_h);
	}
#endif
	while(!_out_queue.empty()) { 
		_out_queue.pop();
	}
	if (_sub != NULL) {
		_sub->removeNetObject(this);
	}
	close(_fd);
}

int netObject::pushMsg(const char *msg)
{
	if (_out_queue.size() < MAX_QUEUE) {
		_out_queue.push(msg);
		
#ifdef U_LOOP
		int ret = sendMsg();
		if(ret == e_ERROR) {
			delete this;
			return ret;
		}
		uloop_fd_add(&_h, _h.flags | ULOOP_WRITE | ULOOP_EDGE_TRIGGER);
#else
		ev_io_stop(_loop, &_h);
		ev_io_init(&_h, evWriteMsg, _h.fd, EV_WRITE);
		ev_io_start(_loop, &_h);
#endif

		return 1;
	}

	ERROR("subsystem %s netObject[%u] out queue size > %d\n", 
				_sub->_name.c_str(), _id, MAX_QUEUE);
	return 0;
}


